// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <nebula/bits/bit_block_counter.h>
#include <algorithm>
#include <cstdint>
#include <type_traits>
#include <nebula/core/buffer.h>
#include <nebula/bits/bitmap_ops.h>

namespace nebula {
namespace internal {

BitBlockCount BitBlockCounter::GetBlockSlow(int64_t block_size) noexcept {
  const int16_t run_length = static_cast<int16_t>(std::min(bits_remaining_, block_size));
  int16_t popcount = static_cast<int16_t>(CountSetBits(bitmap_, offset_, run_length));
  bits_remaining_ -= run_length;
  // This code path should trigger _at most_ 2 times. In the "two times"
  // case, the first time the run length will be a multiple of 8 by construction
  bitmap_ += run_length / 8;
  return {run_length, popcount};
}

OptionalBitBlockCounter::OptionalBitBlockCounter(const uint8_t* validity_bitmap,
                                                 int64_t offset, int64_t length)
    : has_bitmap_(validity_bitmap != nullptr),
      position_(0),
      length_(length),
      counter_(turbo::make_non_null(validity_bitmap), offset, length) {}

OptionalBitBlockCounter::OptionalBitBlockCounter(
    const std::shared_ptr<Buffer>& validity_bitmap, int64_t offset, int64_t length)
    : OptionalBitBlockCounter(validity_bitmap ? validity_bitmap->data() : nullptr, offset,
                              length) {}

OptionalBinaryBitBlockCounter::OptionalBinaryBitBlockCounter(const uint8_t* left_bitmap,
                                                             int64_t left_offset,
                                                             const uint8_t* right_bitmap,
                                                             int64_t right_offset,
                                                             int64_t length)
    : has_bitmap_(HasBitmapFromBitmaps(left_bitmap != nullptr, right_bitmap != nullptr)),
      position_(0),
      length_(length),
      unary_counter_(
          turbo::make_non_null(left_bitmap != nullptr ? left_bitmap : right_bitmap),
          left_bitmap != nullptr ? left_offset : right_offset, length),
      binary_counter_(turbo::make_non_null(left_bitmap), left_offset,
                      turbo::make_non_null(right_bitmap), right_offset, length) {}

OptionalBinaryBitBlockCounter::OptionalBinaryBitBlockCounter(
    const std::shared_ptr<Buffer>& left_bitmap, int64_t left_offset,
    const std::shared_ptr<Buffer>& right_bitmap, int64_t right_offset, int64_t length)
    : OptionalBinaryBitBlockCounter(
          left_bitmap ? left_bitmap->data() : nullptr, left_offset,
          right_bitmap ? right_bitmap->data() : nullptr, right_offset, length) {}

}  // namespace internal
}  // namespace nebula
